package com.wei.violet.app

import com.wei.violet.config.ConfigHandle
import org.apache.log4j.{Level, Logger}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.{DataFrame, Row, SQLContext, SaveMode}

object SparkCoreAllHandler {


  Logger.getLogger("org").setLevel(Level.WARN)

  def main(args: Array[String]): Unit = {

    val conf: SparkConf = new SparkConf()
      .setAppName("将统计结果添加到mysql数据库中")
      .setMaster("local[*]")
      .set("spark.serizlizer", "org.apache.spark.serializer.KryoSerializer")

    val sc = new SparkContext(conf)
    val sQLContext = new SQLContext(sc)
    val dataFrame: DataFrame = sQLContext.read.parquet("F:\\bigdataFile\\parquet")
    val basedata = dataFrame.map(d => {
      val requestmode = d.getAs[Int]("requestmode")
      val processnode = d.getAs[Int]("processnode")
      val iseffective = d.getAs[Int]("iseffective")
      val isbilling = d.getAs[Int]("isbilling")
      val isbid = d.getAs[Int]("isbid")
      val iswin = d.getAs[Int]("iswin")
      val adorderid = d.getAs[Int]("adorderid")
      val provincename = d.getAs[String]("provincename")
      val cityname = d.getAs[String]("cityname")
      val winprice = d.getAs[Double]("winprice")
      val adpayment = d.getAs[Double]("adpayment")
      val ispname=d.getAs[String]("ispname")
      val networkmannername=d.getAs[String]("networkmannername")

      val devicetypeflag=d.getAs[Int]("devicetype")
      val devicetype=if(devicetypeflag==1){"手机"}else if(devicetypeflag==2){"平板"}else{"其他"}
      val clientflag=d.getAs[Int]("client")
      val client=if(clientflag==1){"android"}else if(clientflag==2){"ios"}else if(clientflag==3){"wp"}else{"其他"}



      var InitialRequest = if (requestmode == 1 && processnode >= 1) 1 else 0
      var EffectiveRequest = if (requestmode == 1 && processnode >= 2) 1 else 0
      var ADRequests = if (requestmode == 1 && processnode == 3) 1 else 0
      var BiddingCount = if (iseffective == 1 && isbilling == 1 && isbid == 1 && adorderid != 0) 1 else 0
      var ADShow = if (requestmode == 2 && iseffective == 1) 1 else 0
      var ADHits = if (requestmode == 3 && iseffective == 1) 1 else 0
      var BiddingSuccess = 0
      var DSPADExpense = 0d
      var DSPADCosting = 0d
      if (iseffective == 1 && isbilling == 1 && iswin == 1) {
        BiddingSuccess = 1
        DSPADExpense = winprice / 1000d
        DSPADCosting = adpayment / 1000d
      }
      ((provincename, cityname,ispname,networkmannername,devicetype,client), List(InitialRequest, EffectiveRequest, ADRequests, BiddingCount, BiddingSuccess, ADShow, ADHits, DSPADExpense, DSPADCosting))
    }).cache()

    val byprovince=basedata.map(d=>((d._1._1,d._1._2),(d._2))).reduceByKey((a, b) => a.zip(b).map(t => t._1 + t._2))
    val byispname=basedata.map(d=>((d._1._3),(d._2))).reduceByKey((a, b) => a.zip(b).map(t => t._1 + t._2))
    val bynetworkmannername=basedata.map(d=>((d._1._4),(d._2))).reduceByKey((a, b) => a.zip(b).map(t => t._1 + t._2))
    val bydevicetype=basedata.map(d=>((d._1._5),(d._2))).reduceByKey((a, b) => a.zip(b).map(t => t._1 + t._2))
    val byclient=basedata.map(d=>((d._1._6),(d._2))).reduceByKey((a, b) => a.zip(b).map(t => t._1 + t._2))
    import sQLContext.implicits._
    val result1 = byprovince.map(x => {
      (x._1._1, x._1._2, x._2(0).toInt, x._2(1).toInt, x._2(2).toInt, x._2(3).toInt, x._2(4).toInt, x._2(5).toInt, x._2(6).toInt, x._2(7), x._2(8))
    }).toDF("provincename", "cityname", "InitialRequest", "EffectiveRequest", "ADRequests", "BiddingCount", "BiddingSuccess", "ADShow", "ADHits", "DSPADExpense", "DSPADCosting")
    result1.write.mode(SaveMode.Overwrite).jdbc(
      ConfigHandle.url,
      ConfigHandle.localtable,
      ConfigHandle.conn
    )

    val result2 = byispname.map(x => {
      (x._1, x._2(0).toInt, x._2(1).toInt, x._2(2).toInt, x._2(3).toInt, x._2(4).toInt, x._2(5).toInt, x._2(6).toInt, x._2(7), x._2(8))
    }).toDF("ispname", "InitialRequest", "EffectiveRequest", "ADRequests", "BiddingCount", "BiddingSuccess", "ADShow", "ADHits", "DSPADExpense", "DSPADCosting")
    result2.write.mode(SaveMode.Overwrite).jdbc(
      ConfigHandle.url,
      "sparkcoreTerminal",
      ConfigHandle.conn
    )

    val result3 = bynetworkmannername.map(x => {
      (x._1, x._2(0).toInt, x._2(1).toInt, x._2(2).toInt, x._2(3).toInt, x._2(4).toInt, x._2(5).toInt, x._2(6).toInt, x._2(7), x._2(8))
    }).toDF("networkmannername", "InitialRequest", "EffectiveRequest", "ADRequests", "BiddingCount", "BiddingSuccess", "ADShow", "ADHits", "DSPADExpense", "DSPADCosting")
    result3.write.mode(SaveMode.Overwrite).jdbc(
      ConfigHandle.url,
      "sparkcoreNetWork",
      ConfigHandle.conn
    )

    val result4 = bydevicetype.map(x => {
      (x._1, x._2(0).toInt, x._2(1).toInt, x._2(2).toInt, x._2(3).toInt, x._2(4).toInt, x._2(5).toInt, x._2(6).toInt, x._2(7), x._2(8))
    }).toDF("devicetype", "InitialRequest", "EffectiveRequest", "ADRequests", "BiddingCount", "BiddingSuccess", "ADShow", "ADHits", "DSPADExpense", "DSPADCosting")
    result4.write.mode(SaveMode.Overwrite).jdbc(
      ConfigHandle.url,
     "sparkcoreEquipment",
      ConfigHandle.conn
    )

    val result5= byclient.map(x => {
      (x._1, x._2(0).toInt, x._2(1).toInt, x._2(2).toInt, x._2(3).toInt, x._2(4).toInt, x._2(5).toInt, x._2(6).toInt, x._2(7), x._2(8))
    }).toDF("client", "InitialRequest", "EffectiveRequest", "ADRequests", "BiddingCount", "BiddingSuccess", "ADShow", "ADHits", "DSPADExpense", "DSPADCosting")
    result5.write.mode(SaveMode.Overwrite).jdbc(
      ConfigHandle.url,
     "sparkcoreSystem",
      ConfigHandle.conn
    )

    sc.stop()
  }

}
